package org.apache.rocketmq.client.java.rpc;

import apache.rocketmq.v2.*;
import com.google.common.util.concurrent.Futures;
import com.google.common.util.concurrent.ListenableFuture;
import com.google.common.util.concurrent.MoreExecutors;
import io.grpc.ClientInterceptor;
import io.grpc.ManagedChannel;
import io.grpc.Metadata;
import io.grpc.netty.shaded.io.grpc.netty.GrpcSslContexts;
import io.grpc.netty.shaded.io.grpc.netty.NettyChannelBuilder;
import io.grpc.netty.shaded.io.netty.handler.ssl.SslContext;
import io.grpc.netty.shaded.io.netty.handler.ssl.SslContextBuilder;
import io.grpc.netty.shaded.io.netty.handler.ssl.util.InsecureTrustManagerFactory;
import io.grpc.stub.MetadataUtils;
import io.grpc.stub.StreamObserver;

import javax.net.ssl.SSLException;
import java.net.InetSocketAddress;
import java.time.Duration;
import java.util.Iterator;
import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.Executor;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.TimeUnit;

import org.apache.rocketmq.client.java.route.Endpoints;

public class RpcClientImpl implements RpcClient{

    //客户端最大空闲时间
    private static final Duration KEEP_ALIVE_DURATION = Duration.ofSeconds(30);

    //grpc最大消息大小
    private static final int GRPC_MAX_MESSAGE_SIZE = Integer.MAX_VALUE;

    //客户端可以访问的服务端地址
    private final Endpoints endpoints;

    //客户端与服务端建立的Channel
    private final ManagedChannel channel;


    //grpc客户端的异步存根，使用这个对象可以向服务端发送请求，这个存根并不支持流式rpc，就是简单的请求——响应模式
    private final MessagingServiceGrpc.MessagingServiceFutureStub futureStub;

    //grpc客户的阻塞存根，使用这个对象可以向服务端发送请求，并且同步等待取服务端的响应
    private final MessagingServiceGrpc.MessagingServiceBlockingStub blockingStub;

    //grpc客户端的异步回调存根，使用这个对象可以处理流式rpc
    private final MessagingServiceGrpc.MessagingServiceStub stub;

    //当前Rpc客户端最新活跃时间
    private long activityNanoTime;

    public RpcClientImpl(Endpoints endpoints) throws SSLException {
        this.endpoints = endpoints;
        final SslContextBuilder builder = GrpcSslContexts.forClient();
        builder.trustManager(InsecureTrustManagerFactory.INSTANCE);
        SslContext sslContext = builder.build();

        final NettyChannelBuilder channelBuilder = NettyChannelBuilder.forTarget(endpoints.getGrpcTarget())
                //设置连接最大空闲时间，经过这个事件之后，客户端要主动PING服务端
                .keepAliveTime(KEEP_ALIVE_DURATION.toNanos(), TimeUnit.NANOSECONDS)
                //设置最大消息大小
                .maxInboundMessageSize(GRPC_MAX_MESSAGE_SIZE)
                //设置日志拦截器
                .intercept(LoggingInterceptor.getInstance())
                .sslContext(sslContext);
        //得到可访问的服务端地址列表
        final List<InetSocketAddress> socketAddresses = endpoints.toSocketAddresses();
        if (null != socketAddresses) {
            //创建IP地址解析器工厂
            final IpNameResolverFactory ipNameResolverFactory = new IpNameResolverFactory(socketAddresses);
            //设置名称解析器工厂
            channelBuilder.nameResolverFactory(ipNameResolverFactory);
        }
        this.channel = channelBuilder.build();
        this.futureStub = MessagingServiceGrpc.newFutureStub(channel);
        this.blockingStub = MessagingServiceGrpc.newBlockingStub(channel);
        this.stub = MessagingServiceGrpc.newStub(channel);
        this.activityNanoTime = System.nanoTime();
    }
    /*
     *@方法描述：包装Future结果的方法，在该方法中会把rpc请求得到的响应封装到RpcInvocation对象中
     */
    private <T> ListenableFuture<RpcInvocation<T>> wrapInvocationContext(ListenableFuture<T> future,
                                                                         Metadata header) {

        return Futures.transformAsync(future, response -> {
            //创建RpcInvocation对象
            final Context context = new Context(endpoints, header);
            //创建RpcInvocation对象，并返回
            return Futures.immediateFuture(new RpcInvocation<>(response, context));
        }, MoreExecutors.directExecutor());
    }
    @Override
    public Duration idleDuration() {
        return Duration.ofNanos(System.nanoTime() - activityNanoTime);
    }

    @Override
    public void shutdown() throws InterruptedException {
        channel.shutdown().awaitTermination(Long.MAX_VALUE, TimeUnit.SECONDS);
    }
    /* @方法描述：向服务端查询路由信息的方法
     */
    @Override
    public ListenableFuture<RpcInvocation<QueryRouteResponse>> queryRoute(Metadata metadata, QueryRouteRequest request, Executor executor, Duration duration) {
        //在这里更新客户端最新活动时间
        this.activityNanoTime = System.nanoTime();
        //下面就是发送路由查询请求的操作，在为futureStub设置了元数据拦截器，把请求元数据附加到请求头，设置了执行操作的执行器，设置了请求超时时间之后，就把请求发送出去了
        //这个操作会返回一个future，当接收到服务端的响应之后，future对象会在grpc内部被设置为完成状态
        final ListenableFuture<QueryRouteResponse> listenableFuture = futureStub.withInterceptors(MetadataUtils.newAttachHeadersInterceptor(metadata))
                .withDeadlineAfter(duration.toNanos(), TimeUnit.NANOSECONDS).withExecutor(executor)
                .queryRoute(request);
        return wrapInvocationContext(listenableFuture, metadata);
    }

    @Override
    public ListenableFuture<RpcInvocation<HeartbeatResponse>> heartbeat(Metadata metadata, HeartbeatRequest request, Executor executor, Duration duration) {
        this.activityNanoTime = System.nanoTime();
        final ListenableFuture<HeartbeatResponse> future =
                futureStub.withInterceptors(MetadataUtils.newAttachHeadersInterceptor(metadata)).withExecutor(executor)
                        .withDeadlineAfter(duration.toNanos(), TimeUnit.NANOSECONDS).heartbeat(request);
        return wrapInvocationContext(future, metadata);
    }


    @Override
    public ListenableFuture<RpcInvocation<SendMessageResponse>> sendMessage(Metadata metadata,
                                                                            SendMessageRequest request, Executor executor, Duration duration) {
        this.activityNanoTime = System.nanoTime();
        final ListenableFuture<SendMessageResponse> future =
                futureStub.withInterceptors(MetadataUtils.newAttachHeadersInterceptor(metadata)).withExecutor(executor)
                        .withDeadlineAfter(duration.toNanos(), TimeUnit.NANOSECONDS).sendMessage(request);
        return wrapInvocationContext(future, metadata);
    }


    @Override
    public ListenableFuture<RpcInvocation<QueryAssignmentResponse>> queryAssignment(Metadata metadata,
                                                                                    QueryAssignmentRequest request, Executor executor, Duration duration) {
        this.activityNanoTime = System.nanoTime();
        final ListenableFuture<QueryAssignmentResponse> future =
                futureStub.withInterceptors(MetadataUtils.newAttachHeadersInterceptor(metadata)).withExecutor(executor)
                        .withDeadlineAfter(duration.toNanos(), TimeUnit.NANOSECONDS).queryAssignment(request);
        return wrapInvocationContext(future, metadata);
    }


    @Override
    public ListenableFuture<RpcInvocation<Iterator<ReceiveMessageResponse>>> receiveMessage(Metadata metadata,
                                                                                            ReceiveMessageRequest request, ExecutorService executor, Duration duration) {
        this.activityNanoTime = System.nanoTime();
        final Callable<Iterator<ReceiveMessageResponse>> callable = () -> blockingStub
                .withInterceptors(MetadataUtils.newAttachHeadersInterceptor(metadata)).withExecutor(executor)
                .withDeadlineAfter(duration.toNanos(), TimeUnit.NANOSECONDS).receiveMessage(request);
        final ListenableFuture<Iterator<ReceiveMessageResponse>> future =
                MoreExecutors.listeningDecorator(executor).submit(callable);
        return wrapInvocationContext(future, metadata);
    }

    @Override
    public ListenableFuture<RpcInvocation<AckMessageResponse>> ackMessage(Metadata metadata,
                                                                          AckMessageRequest request, Executor executor, Duration duration) {
        this.activityNanoTime = System.nanoTime();
        final ListenableFuture<AckMessageResponse> future =
                futureStub.withInterceptors(MetadataUtils.newAttachHeadersInterceptor(metadata)).withExecutor(executor)
                        .withDeadlineAfter(duration.toNanos(), TimeUnit.NANOSECONDS).ackMessage(request);
        return wrapInvocationContext(future, metadata);
    }



    @Override
    public ListenableFuture<RpcInvocation<ChangeInvisibleDurationResponse>> changeInvisibleDuration(Metadata metadata, ChangeInvisibleDurationRequest request, Executor executor, Duration duration) {
        return null;
    }

    @Override
    public ListenableFuture<RpcInvocation<ForwardMessageToDeadLetterQueueResponse>> forwardMessageToDeadLetterQueue(Metadata metadata, ForwardMessageToDeadLetterQueueRequest request, Executor executor, Duration duration) {
        return null;
    }

    @Override
    public ListenableFuture<RpcInvocation<EndTransactionResponse>> endTransaction(Metadata metadata, EndTransactionRequest request, Executor executor, Duration duration) {
        return null;
    }

    @Override
    public ListenableFuture<RpcInvocation<NotifyClientTerminationResponse>> notifyClientTermination(Metadata metadata, NotifyClientTerminationRequest request, Executor executor, Duration duration) {
        return null;
    }

    @Override
    public StreamObserver<TelemetryCommand> telemetry(Metadata metadata, Executor executor, Duration duration,
                                                      StreamObserver<TelemetryCommand> responseObserver) {
        final ClientInterceptor interceptor = MetadataUtils.newAttachHeadersInterceptor(metadata);
        final MessagingServiceGrpc.MessagingServiceStub stub0 = this.stub.withInterceptors(interceptor)
                .withExecutor(executor).withDeadlineAfter(duration.toNanos(), TimeUnit.NANOSECONDS);
        return stub0.telemetry(responseObserver);
    }
}
